package com.mls.guardian.remoting.netty;

import com.mls.guardian.common.RemoteCommandType;
import com.mls.guardian.config.NettyConfig;
import com.mls.guardian.protocol.CodeType;
import com.mls.guardian.protocol.RemoteCommand;
import com.mls.guardian.protocol.body.ClientInactiveBody;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by jl on 15/9/9.
 */
public class NettyServer extends AbstractNetty{
    private static final Logger log = LoggerFactory.getLogger(NettyServer.class);
    private int port;
    private ServerBootstrap bootstrap=new ServerBootstrap();
    private ConcurrentHashMap<String,Channel> clientChannel = new ConcurrentHashMap<String,Channel>();
    private final EventLoopGroup eventLoopGroupWorker;
    private final EventLoopGroup eventLoopGroupBoss;
    private final EventExecutorGroup eventExecutors;
    // 定时器
    private final Timer timer = new Timer("scanResponseService", true);

    public NettyServer(int port,NettyConfig nettyConfig){
        super(nettyConfig);
        this.port = port;

        this.eventExecutors = new DefaultEventExecutorGroup(10,new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);


            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r,
                        String.format("NettyExecutor_%d", this.threadIndex.incrementAndGet()));
            }
        });

        this.eventLoopGroupBoss = new NioEventLoopGroup(3, new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);


            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r,
                        String.format("NettyBossSelector_%d", this.threadIndex.incrementAndGet()));
            }
        });

        this.eventLoopGroupWorker =
                new NioEventLoopGroup(nettyConfig.getMasterWorkerNum(), new ThreadFactory() {
                    private AtomicInteger threadIndex = new AtomicInteger(0);
                    private int threadTotal = 10;


                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, String.format("NettyServerSelector_%d_%d", threadTotal,
                                this.threadIndex.incrementAndGet()));
                    }
                });
    }

    public void start() throws InterruptedException {
        bootstrap.group(eventLoopGroupBoss,eventLoopGroupWorker);
        bootstrap.channel(NioServerSocketChannel.class);
        bootstrap.option(ChannelOption.SO_BACKLOG, 128);
        //通过NoDelay禁用Nagle,使消息立即发出去，不用等待到一定的数据量才发出去
        bootstrap.option(ChannelOption.TCP_NODELAY, true);
        //保持长连接状态
        bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);

        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                ChannelPipeline p = socketChannel.pipeline();
                p.addLast(new NettyDecoder());
                p.addLast(new NettyEncoder());
                p.addLast(eventExecutors,new NettyServerHandler());
            }
        });

        ChannelFuture f= bootstrap.bind(port).sync();
        if(f.isSuccess()){
            log.info("netty server start---------------");

            // 每隔1秒扫描下异步调用超时情况
            this.timer.scheduleAtFixedRate(new TimerTask() {

                @Override
                public void run() {
                    try {
                        NettyServer.this.scanResponseTable();
                    }
                    catch (Exception e) {
                        log.error("scanResponseTable exception", e);
                    }
                }
            }, 1000 * 3, 1000);

        }else{
            throw new RuntimeException("启动nettyserver出错");
        }
    }



    private class NettyServerHandler extends SimpleChannelInboundHandler<RemoteCommand>{
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RemoteCommand request) throws Exception {
            handleReciveMessage(ctx,request);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            RemoteCommand remoteCommand = new RemoteCommand();
            ClientInactiveBody clientInactive = new ClientInactiveBody();
            clientInactive.setAddr(RemoteTools.getRemoteAddr(ctx.channel()));
            remoteCommand.setRemoteCommandType(RemoteCommandType.REQUEST);
            remoteCommand.setCode(CodeType.CLIENT_INVALID);
            remoteCommand.setRemoteBody(clientInactive);

            handleReciveMessage(ctx,remoteCommand);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            cause.printStackTrace();
            ctx.fireExceptionCaught(cause);
        }
    }


    @Override
    public  void sendMessageOneWay(Channel channel,final RemoteCommand request){
        channel.writeAndFlush(request);
    }

    @Override
    public void sendMessageOneWay(String addr, RemoteCommand request) throws Exception {

    }


    public void shutdown(){
        if (this.timer != null) {
            this.timer.cancel();
        }

        this.eventLoopGroupBoss.shutdownGracefully();

        this.eventLoopGroupWorker.shutdownGracefully();

        this.eventExecutors.shutdownGracefully();

        clientChannel.clear();
    }
}
